#include <util/list_table.h>
#include "sf_atomic.h"
#include "sf_mutex.h"
#include "common_log.h"
#include <map>
#include "dispatch_process.h"

#define MPMAX(a, b) ((a) > (b) ? (a) : (b))
#define MPMIN(a, b) ((a) > (b) ? (b) : (a))

#ifndef USE_DESTRUC_OBJ
#define USE_DESTRUC_OBJ
#else
#undef USE_DESTRUC_OBJ
#endif

namespace dispatch {

#define SINGLE_TIMER  (Timer *)(0x1)

inline static void getRealtime(struct timespec *out_ts)
{
    clock_gettime(CLOCK_MONOTONIC, out_ts);
}

inline static int64_t timeUs(void)
{
    struct timespec ts;
    SF_ASSERT(0 == clock_gettime(CLOCK_MONOTONIC, &ts));
    return (int64_t)ts.tv_sec * 1000000LL + ts.tv_nsec / 1000;
}

static struct timespec timeUsToTimespec(int64_t time_us)
{
    struct timespec ts;
    getRealtime(&ts);

    int64_t diff_us    = time_us;
    int64_t diff_secs  = diff_us / (1000L * 1000L);
    int64_t diff_nsecs = (diff_us - diff_secs * (1000L * 1000L)) * 1000L;

    if (diff_nsecs < 0)
    {
        diff_secs  -= 1;
        diff_nsecs += 1000000000L;
    }

    if ((int64_t)(diff_nsecs + ts.tv_nsec) >= 1000000000LL)
    {
        diff_secs  += 1;
        diff_nsecs -= 1000000000UL;
    }

    // OSX can't deal with large timeouts. Also handles tv_sec/time_t overflows.
    diff_secs   = MPMIN(diff_secs, 10000000);
    ts.tv_sec  += diff_secs;
    ts.tv_nsec += diff_nsecs;

    return ts;
}

//! 参与调度执行数据结构体
class DispatchItem
{
public:
    DispatchItem() {}
    DispatchItem(Object::Callback1 fn, Object *data,
                 bool async = false,
                 bool merge = false,
                 bool compe = false)
        : asynchronous(async), mergeable(merge)
        , completed(compe), delobject(false)
        , fn(fn), fn_data(data)
        , next(NULL)
    {}

    bool              asynchronous;
    bool              mergeable; // 合并相同的消息
    bool              completed;
    bool              delobject;

    Object::Callback1 fn;
    Object           *fn_data;
    DispatchItem     *next;
};

class LoopEvtItem;
class ItemTimer : public DispatchItem
{
public:
    ItemTimer() {}
    ItemTimer(Object::Callback1 fn, Object *fn_data, bool = 0, bool = 0)
        : DispatchItem(fn, fn_data, true)
        , that(NULL), nodeHandler(NULL), tiggerTime(0)
    {}

    Timer       *that;
    LoopEvtItem *nodeHandler;
    int64_t      tiggerTime; // (us)下一次触发的时间，是实时时间(微妙/us)
};

class LoopEvtItem
{
public:
    Loop::Evt    evt;
    LoopEvtItem *next;
};

#if 0
class NodePool2
{
public:
    enum QueueStatus
    {
        FREE = 0,
        PREPARE = 1,
    };
    class Node
    {
    public:
        template<typename T> T* get() { return reinterpret_cast<T*> (a); }
        char a[sizeof(ItemTimer)];
    };

    typedef Node _NodeType;

    NodePool2(int node_num = 30)
    {
        _NodeType tmp;
        for(int i = 0; i < node_num; ++i)
        {
            mQu[FREE].push_back(tmp);
        }
    }

    template<typename T>
    T *getNode(Object::Callback1 fn, Object *data, bool async = false, bool merge = false)
    {
        Node *nd = getFreeNode();
        if (UNLIKELY(!nd)) return NULL;

        T *item = nd->get<T>();
        *item   = T(fn, data, async, merge);
        return item;
    }

    LoopEvtItem *getLoopNode(const Loop::Evt &e)
    {
        Node *nd = getFreeNode();
        if (UNLIKELY(!nd)) return NULL;
        LoopEvtItem *item = nd->get<LoopEvtItem>();
        item->evt = e;
        item->next= NULL;
        return item;
    }

    void recycle(void *item)
    {
        putNode(FREE, reinterpret_cast<_NodeType *>(item));
    }

    bool empty(QueueStatus type)
    {
        Autolock l(&mMtx);
        return mQu[type].empty();
    }

    _NodeType *getFreeNode()
    {
        Autolock l(&mMtx);
        typename ListTable<_NodeType>::iterator it = mQu[FREE].pop_front();
        return (it != mQu[FREE].end()) ? &(*it) : 0;
    }

    _NodeType *getPrepareNode()
    {
        Autolock l(&mMtx);
        typename ListTable<_NodeType>::iterator it = mQu[PREPARE].pop_front();
        return (it != mQu[PREPARE].end()) ? &(*it) : nullptr;
    }

    /**
     * @brief putNode
     * @param type [in] node放入那个队列
     * @param n [in] 数据节点
     * @warning n必须是getNode的返回值，否则系统出错，出现指针乱指
     * @details 巧妙的运用，数据块和连接(next/prev)位置的相对位置，找到ListNode正确的指针
     */
    void putNode(QueueStatus type, _NodeType *n)
    {
        Autolock l(&mMtx);
        //相当是把n的指针前移sizeof(ListNode)位置，找到正确的ListNode的指针
        typename ListTable<_NodeType>::iterator it((typename ListTable<_NodeType>::ListItem *)((char *)n - sizeof(ListNode)));
        mQu[type].push_back(it);
    }

    void profile()
    {
        Autolock l(&mMtx);
        GEN_Printf(LOG_DEBUG, "used=%d, free=%d"
                   , mQu[PREPARE].size(), mQu[FREE].size());
    }

private:
    NodePool2(NodePool2 &);
    NodePool2& operator=(NodePool2 &);

    ListTable<Node> mQu[2];
    SFMutex         mMtx;
};
#endif

//#define USE_MALLOC 1
/**
 * @brief The NodePool class
 * @note node事件池子
 */
class NodePool
{
public:
    NodePool()
        : mCount(0), mUsed(0)
    {
        //GEN_Printf(LOG_DEBUG, "sizeof(ItemTimer): %lu", sizeof(ItemTimer));
        //GEN_Printf(LOG_DEBUG, "sizeof(LoopEvtItem): %lu", sizeof(LoopEvtItem));
    }

    class Node
    {
    public:
        template<typename T> T* get() { return reinterpret_cast<T*> (a); }
        char a[sizeof(ItemTimer)];
    };

    template<typename T>
    T *getNode(Object::Callback1 fn, Object *data, bool async = false, bool merge = false)
    {
#ifndef USE_MALLOC
        Autolock l(&mMtx);
        if (mPools.empty())
        {
            mPools.push_back(Node());
            ++mCount;
        }

        ListTable<Node>::iterator it = mPools.pop_back();
#else
        Node *it = new Node;
#endif
        T *item = it->get<T>();
        *item   = T(fn, data, async, merge);

        ++mUsed;
        //GEN_Printf(LOG_DUMP, "GET: %p", item);

        return item;
    }

    LoopEvtItem *getLoopNode(const Loop::Evt &e)
    {
#ifndef USE_MALLOC
        Autolock l(&mMtx);
        if (mPools.empty())
        {
            mPools.push_back(Node());
            ++mCount;
        }

        ListTable<Node>::iterator it = mPools.pop_back();
#else
        Node *it = new Node;
#endif
        LoopEvtItem *item = it->get<LoopEvtItem>();

        item->evt = e;
        item->next= NULL;

        ++mUsed;
        //GEN_Printf(LOG_DUMP, "GET: %p", item);

        return item;
    }

    void recycle(void *item)
    {
#ifndef USE_MALLOC
        Autolock l(&mMtx);
        ListTable<Node>::iterator it((ListTable<Node>::ListItem *)((char *)item - sizeof(ListNode)));
        mPools.push_back(it);
        --mUsed;
        //GEN_Printf(LOG_DUMP, "PUT: %p", item);
#else
        //GEN_Printf(LOG_DUMP, "PUT: %p", item);
        Node *it = (Node*)item;
        delete it;
#endif
    }

    void profile()
    {
        GEN_Printf(LOG_INFO, "node pools: used=%d, counts=%d", mUsed, mCount);
    }

private:
    int             mCount;
    int             mUsed;
    SFMutex         mMtx;
    ListTable<Node> mPools;
};

static NodePool *gPool = NULL;
class QueuePrivate : public Object
{
    SFAtomic<uint> refCount;
public:
    QueuePrivate()
        : refCount(1)
        , head(NULL), tail(NULL)
        , timerhead(NULL)
        , loopHead(NULL), loopTail(NULL)
        , wakeup_fn(NULL), wakeup_ctx(NULL)
        , wait(0), interrupted(false), in_process(false)
        , in_process_thread(0), locked(false)
        , lock_requests(0), locked_explicit(false)
        , locked_explicit_thread(0)
        , timerProc(NULL), exitLoop(false)
    {
        //GEN_Printf(LOG_DEBUG, "this: %p", this);
#if !defined (__MACH__) && !defined (__APPLE__)
        pthread_condattr_t  condattr;

        SF_ASSERT(0 ==  pthread_condattr_init(&condattr));
        pthread_condattr_setclock(&condattr, CLOCK_MONOTONIC);

        pthread_mutex_init(&lock, NULL);
        pthread_cond_init(&cond, &condattr);

        pthread_condattr_destroy(&condattr);
#else
        pthread_mutex_init(&lock, NULL);
        pthread_cond_init(&cond, NULL);
#endif

        if (!gPool)
        {
            gPool = new NodePool;
        }
    }

    QueuePrivate *ref()
    {       
        refCount.inc();
//        ++refCount;
        return this;
    }
    void unref()
    {
        if (0 == refCount.dec())
//        if (--refCount == 0)
        {
            GEN_Printf(LOG_INFO, "release queue private.");
            delete this;
        }
    }

    void append(DispatchItem *item);
    void appendTimer(Timer *fn_data, int usec);
    void remove(Callback1 fn, Object *fn_data);
    void removeTimer(Timer *timer);    
    ItemTimer *findTimer(Timer *timer);

    void _appendTimer(ItemTimer *item);
    int64_t getWait();
    bool haveTigger()
    {
        return timerhead ? (timeUs() > timerhead->tiggerTime ? true : false) : false;
    }

    void procTimer();
    void procEventFunc();

    void wakeup()
    {
        // 表示在同一执行流程下，是不需要唤醒的
        // pthread_equal==0两线程ID不相等，pthread_equal!=0两线程ID相等
        if (wakeup_ctx && (0 == pthread_equal(in_process_thread, pthread_self())))
        {
            (wakeup_ctx->*wakeup_fn)();
        }
    }
    void interrupt();
    void runIdle();

public:
    DispatchItem *head, *tail;
    ItemTimer    *timerhead;  // 定时器运行队列
    LoopEvtItem  *loopHead, *loopTail;

    pthread_mutex_t lock;
    pthread_cond_t  cond;

    Callback1  wakeup_fn;
    Object    *wakeup_ctx;

    // Time at which mp_dispatch_queue_process() should return.
    int64_t wait;
    // Make mp_dispatch_queue_process() exit if it's idle.
    bool interrupted;
    // The target thread is in mp_dispatch_queue_process() (and either idling,
    // locked, or running a dispatch callback).
    bool in_process;
    pthread_t in_process_thread;
    // The target thread is in mp_dispatch_queue_process(), and currently
    // something has exclusive access to it (e.g. running a dispatch callback,
    // or a different thread got it with mp_dispatch_lock()).
    bool locked;
    // A mp_dispatch_lock() call is requesting an exclusive lock.
    size_t lock_requests;
    // locked==true is due to a mp_dispatch_lock() call (for debugging).
    bool locked_explicit;
    pthread_t locked_explicit_thread;
    // 定时器处理期间
    Timer *timerProc;

    // 事件循环是否退出的标志
    bool exitLoop;
private:
    ~QueuePrivate()
    {
        SF_ASSERT(!head);
        SF_ASSERT(!in_process);
        SF_ASSERT(!lock_requests);
        SF_ASSERT(!locked);
        ItemTimer **pcur = &timerhead;
        while (*pcur)
        {
            ItemTimer *cur = static_cast<ItemTimer *>(*pcur);
            *pcur = static_cast<ItemTimer *>(cur->next);
            gPool->recycle(cur);
        }
        pthread_cond_destroy(&cond);
        pthread_mutex_destroy(&lock);
    }
};

Queue::Queue()
    : d(new QueuePrivate)
{
}

Queue::~Queue()
{
    d->unref();
}

int64_t Queue::getNeedWait()
{
    int64_t tmp;

    pthread_mutex_lock(&d->lock);
    tmp = d->getWait();
    pthread_mutex_unlock(&d->lock);

    return tmp;
}

void Queue::setWakeupFn(Callback1 wakeup_fn, Object *ctx)
{
    d->wakeup_fn  = wakeup_fn;
    d->wakeup_ctx = ctx;
}

void Queue::enqueue(Callback1 fn, Object *fn_data)
{
    SF_ASSERT(fn_data);
    DispatchItem *item = gPool->getNode<DispatchItem>(fn, fn_data, true);

    d->append(item);
}

void Queue::enqueueAutofree(Callback1 fn, Object *fn_data)
{
#ifdef USE_DESTRUC_OBJ
    SF_ASSERT(fn_data);
    DispatchItem *item = gPool->getNode<DispatchItem>(fn, fn_data, true);
    item->delobject = true;

    d->append(item);
#else
    UNUSED(fn);
    UNUSED(fn_data);
    GEN_Printf(LOG_ERROR, "Qbject virtual ~Qbject not defined.");
    SF_ASSERT(0);
#endif // USE_DESTRUC_OBJ
}

void Queue::enqueueNotify(Callback1 fn, Object *fn_data)
{
    SF_ASSERT(fn_data);
    DispatchItem *item  = gPool->getNode<DispatchItem>(fn, fn_data, true, true);

    d->append(item);
}

void Queue::cancelFn(Callback1 fn, Object *fn_data)
{
    SF_ASSERT(fn_data);
    pthread_mutex_lock(&d->lock);

    d->remove(fn, fn_data);

    pthread_mutex_unlock(&d->lock);
}

void Queue::run(Callback1 fn, Object *fn_data)
{
    SF_ASSERT(fn_data);
    DispatchItem item(fn, fn_data);
    d->append(&item);

    pthread_mutex_lock(&d->lock);
    while (!item.completed)
    {
        pthread_cond_wait(&d->cond, &d->lock);
    }
    pthread_mutex_unlock(&d->lock);
}

void Queue::queueProcess(int timeout)
{
    bool needWait = (timeout != 0) ? true : false;
    pthread_mutex_lock(&d->lock);

    // d->wait = timeout > 0 ? timeout * 1000000LL : 0;
    SF_ASSERT(!d->in_process); // recursion not allowed
    d->in_process = true;
    d->in_process_thread = pthread_self();

    // Wake up thread which called lock().
    if (d->lock_requests)
    {
        pthread_cond_broadcast(&d->cond);
    }

    while (1)
    {
        //GEN_Printf(LOG_DEBUG, "loop start");
        if (d->lock_requests)
        {
            // Block due to something having called lock().
            //GEN_Printf(LOG_DEBUG, "lock wait start");
            pthread_cond_wait(&d->cond, &d->lock);
            //GEN_Printf(LOG_DEBUG, "lock wait end");
        } else if (d->head)
        {
            d->procEventFunc();
        } else if (d->haveTigger())
        {
            d->procTimer();
        }
#if 0
        else if (d->wait > 0 && !d->interrupted)
        {
            struct timespec ts = timeUsToTimespec(d->wait);
            if (pthread_cond_timedwait(&d->cond, &d->lock, &ts))
            {
                d->wait = 0;
            }
        }
#else
        else if (!d->interrupted && needWait)
        {
            if ( timeout > 0)
            {
                struct timespec ts = timeUsToTimespec(timeout * 1000);
                pthread_cond_timedwait(&d->cond, &d->lock, &ts);
                needWait = false;
                // 下一个循环退出，不能直接break中断循环，
                // 如果外部正在调用lock，这个变量locked会至为true，那么就会触发下面的断言
                // break;
            } else
            {
                d->wait = d->getWait();

                // GEN_Printf(LOG_DEBUG, "loop wait start, wait=%ld", d->wait);
                if (d->wait > 0)
                {
                    struct timespec ts = timeUsToTimespec(d->wait);
                    pthread_cond_timedwait(&d->cond, &d->lock, &ts);
                }
            }
            //GEN_Printf(LOG_DEBUG, "loop wait start");
        }
#endif
        else
        {
            break;
        }
    }

    SF_ASSERT(!d->locked);
    d->in_process  = false;
    d->interrupted = false;
    pthread_mutex_unlock(&d->lock);
}

void Queue::interrupt()
{
    pthread_mutex_lock(&d->lock);
    d->interrupted = true;
    pthread_cond_broadcast(&d->cond);
    pthread_mutex_unlock(&d->lock);
}

void Queue::adjustTimeout(long until)
{
    pthread_mutex_lock(&d->lock);
    if (d->in_process && d->wait > until)
    {
        d->wait = until;
        pthread_cond_broadcast(&d->cond);
    }
    pthread_mutex_unlock(&d->lock);
}

void Queue::lock()
{
    pthread_mutex_lock(&d->lock);
    // 绝不能从分派的回调函数中递归地调用。
    if (d->in_process)
    {
        // pthread_equal==0两线程ID不相等，pthread_equal!=0两线程ID相等
        SF_ASSERT(!pthread_equal(d->in_process_thread, pthread_self()));
    }
    // 绝对不能被递归地调用。
    if (d->locked_explicit)
    {
        SF_ASSERT(!pthread_equal(d->locked_explicit_thread, pthread_self()));
    }
    d->lock_requests += 1;

    // 现在等待，直到目标线程被“困住”在queueProcess()调用中，这意味着我们可以独占访问目标线程状态。
    while (!d->in_process)
    {
        pthread_mutex_unlock(&d->lock);

        d->wakeup();

        pthread_mutex_lock(&d->lock);
        if (d->in_process)
        {
            break;
        }
        pthread_cond_wait(&d->cond, &d->lock);
    }
    // Wait until we can get the lock.
    while (!d->in_process || d->locked)
    {
        pthread_cond_wait(&d->cond, &d->lock);
    }
    // "Lock".
    SF_ASSERT(d->lock_requests);
    SF_ASSERT(!d->locked);
    SF_ASSERT(!d->locked_explicit);
    d->locked = true;
    d->locked_explicit = true;
    d->locked_explicit_thread = pthread_self();
    pthread_mutex_unlock(&d->lock);
}

void Queue::unlock()
{
    pthread_mutex_lock(&d->lock);
    SF_ASSERT(d->locked);
    // Must be called after a lock(), from the same thread.
    SF_ASSERT(d->locked_explicit);
    SF_ASSERT(pthread_equal(d->locked_explicit_thread, pthread_self()));
    // "Unlock".
    d->locked = false;
    d->locked_explicit = false;
    d->lock_requests -= 1;
    // 唤醒queueProcess()，或者其他lock()。(如果lock_requests>0，只唤醒1个其他locker就好了。)
    pthread_cond_broadcast(&d->cond);
    pthread_mutex_unlock(&d->lock);
}

bool Queue::isLocked()
{
    bool is;

    pthread_mutex_lock(&d->lock);
    is = d->locked;
    pthread_mutex_unlock(&d->lock);

    return is;
}

void Queue::profile()
{
    GEN_Printf(LOG_DUMP, "-----------debug info begin: %p---------------", d);
    gPool->profile();

    pthread_mutex_lock(&d->lock);
    GEN_Printf(LOG_DEBUG, "curtime: %lld, timerhead: %p", (long long)timeUs(), d->timerhead);

    if (d->timerhead)
    {
        ItemTimer *cur = d->timerhead;
        while (cur)
        {
            GEN_Printf(LOG_DEBUG, "Timer[%p] Info, tigger: %lld, handler: %p, interval: %d, repeat: %d", cur,
                       (long long)d->timerhead->tiggerTime, cur->that, cur->that->mMSec, cur->that->mRepeat);
            cur = (ItemTimer *)cur->next;
        }
    }

    GEN_Printf(LOG_DEBUG, "haveTigger: %d", d->haveTigger());

    {
        GEN_Printf(LOG_DEBUG, "loop: %p, %p", d->loopHead, d->loopTail);
        LoopEvtItem *cur = d->loopHead;
        while (cur) {
            GEN_Printf(LOG_DEBUG, "EVT: sig: %d, wParam: %ld", cur->evt.sig, cur->evt.wParam);
            cur = cur->next;
        }
    }

    {
        GEN_Printf(LOG_DEBUG, "func: %p, %p", d->head, d->tail);
        DispatchItem *cur = d->head;
        while (cur) {
            GEN_Printf(LOG_DEBUG, "FUNC: fn_data: %p", cur->fn_data);
            cur = cur->next;
        }
    }

    pthread_mutex_unlock(&d->lock);
    GEN_Printf(LOG_DUMP, "-----------debug info end---------------");
}

bool Queue::isCommonFlow()
{
    return 0 == pthread_equal(d->in_process_thread, pthread_self()) ? true : false;
}

void QueuePrivate::append(DispatchItem *item)
{
    pthread_mutex_lock(&lock);
    if (item->mergeable)
    {
        for (DispatchItem *cur = head; cur; cur = cur->next)
        {
            if (cur->mergeable && cur->fn == item->fn &&
                cur->fn_data == item->fn_data)
            {
                gPool->recycle(item);
                pthread_mutex_unlock(&lock);
                return;
            }
        }
    }

    if (tail)
    {
        tail->next = item;
    } else
    {
        head = item;
    }
    tail = item;

    // Wake up the main thread; note that other threads might wait on this
    // condition for reasons, so broadcast the condition.
    pthread_cond_broadcast(&cond);
    // No wakeup callback -> assume queueProcess() needs to be
    // interrupted instead.
    if (!wakeup_fn)
    {
        interrupted = true;
    }
    pthread_mutex_unlock(&lock);

    wakeup();
}

void QueuePrivate::appendTimer(Timer *that, int usec)
{
    ItemTimer *item  = gPool->getNode<ItemTimer>(that->fn, that->ctx, true);
    item->tiggerTime = timeUs() + usec;
    item->that       = that;

    // GEN_Printf(LOG_DUMP, "appendTimer: %p -- %p", item->that, that);
    _appendTimer(item);
}

void QueuePrivate::remove(Object::Callback1 fn, Object *fn_data)
{
    DispatchItem **pcur = &head;
    tail = NULL;
    while (*pcur)
    {
        DispatchItem *cur = *pcur;
        if (cur->fn == fn && cur->fn_data == fn_data)
        {
            *pcur = cur->next;
            gPool->recycle(cur);
        } else
        {
            tail = cur;
            pcur = &cur->next;
        }
    }
}

void QueuePrivate::removeTimer(Timer *timer)
{
    ItemTimer **pcur = &timerhead;

    while (*pcur)
    {
        ItemTimer *cur = *pcur;
        if (cur->that == timer)
        {
            *pcur = static_cast<ItemTimer *>(cur->next);
            gPool->recycle(cur);
        } else
        {
            pcur = (ItemTimer **)&cur->next;
        }
    }
}

//bool QueuePrivate::findTimer(Timer *timer)
//{
//    ItemTimer *cur = timerhead;
//    while (cur)
//    {
//        if (cur->that == timer) return true;
//        cur = (ItemTimer *)cur->next;
//    }

//    return false;
//}

ItemTimer *QueuePrivate::findTimer(Timer *timer)
{
    ItemTimer **pcur = &timerhead;

    while (*pcur)
    {
        ItemTimer *cur = *pcur;
        if (cur->that == timer)
        {
            *pcur = static_cast<ItemTimer *>(cur->next);
            cur->next = nullptr;
            return (ItemTimer *)cur;
        } else
        {
            pcur = (ItemTimer **)&cur->next;
        }
    }
    return nullptr;
}

void QueuePrivate::_appendTimer(ItemTimer *item)
{
    //GEN_Printf(LOG_INFO, "_appendTimer: item=%p, item->that=%p, tiggerTime: %lld, mSec: %d", item, item->that, item->tiggerTime, item->that->mMSec);
    if (timerhead)
    {
        ItemTimer **pcur = &timerhead;

        while (1)
        {
            ItemTimer *cur = *pcur;

            // GEN_Printf(LOG_INFO, "cur:  item=%p, item->that=%p, tiggerTime: %lld, nodeHandler: %p", cur, cur->that, cur->tiggerTime, cur->nodeHandler);
            // tiggerTime 都表示的是，下一次定时器触发的时间
            if (cur->tiggerTime > item->tiggerTime)
            {
                item->next = cur;
                *pcur = item;
                break;
            }

            // 说明当前节点已经是在末尾了, 就把ITEM加到最后位置
            if (nullptr == cur->next)
            {
//                GEN_Printf(LOG_INFO, "cur:  item=%p, item->that=%p, tiggerTime: %lld", cur, cur->that, cur->tiggerTime);
//                GEN_Printf(LOG_INFO, "item: item=%p, item->that=%p, tiggerTime: %lld", item, item->that, item->tiggerTime);
                cur->next  = item;
                item->next = nullptr;
                break;
            } else
            {
                pcur = (ItemTimer **)&cur->next;
            }
        }
    } else
    {
        timerhead = item;
    }
}

int64_t QueuePrivate::getWait()
{
    int64_t curTime = 0;
    if (timerhead)
    {
        // 只要定时器队列存在节点，必然就是在运行的，没有运行的必然不再这个队列中
        ItemTimer *cur = timerhead;
        int64_t  curUs = timeUs();
        // TODO: warnning 如果后续事件循环，有延时了tiggerTime会小于当前时间，此时需要强制设置为0，不能返回负数
        // 否则后续，处理负数时，会出现各种异常问题
        //GEN_Printf(LOG_DEBUG, "getWait: %ld - %ld", cur->tiggerTime, timeUs());
        curTime = cur->tiggerTime > curUs ? cur->tiggerTime - curUs : 0;
    } else
    {
        curTime = INT64_MAX;
    }

    return curTime;
}

void QueuePrivate::procTimer()
{
    // 只是处理第一个节点, 因为整个列表都是按照时间循序插入的
    if (timerhead)
    {
        ItemTimer *cur = timerhead;

        // 32位 和 64位，传入的值，根据指定的%lld 还是%ld，会导致下面的数据出现错误
        //GEN_Printf(LOG_DEBUG, "tiggerTime cur diff: %lld", timeUs() - cur->tiggerTime);
        if (cur->tiggerTime <= timeUs())
        {
            SF_ASSERT(NULL == timerProc);
            timerProc = cur->that;
            locked    = true;
            timerhead = static_cast<ItemTimer *>(cur->next);

            pthread_mutex_unlock(&lock);

            //GEN_Printf(LOG_DEBUG, "cur: %p, %p, %p, timerProc: %p, start", cur, cur->fn_data, this, timerProc);
            (cur->fn_data->*(cur->fn))();
            //GEN_Printf(LOG_DEBUG, "cur: %p, timerProc: %p, end", cur, timerProc);

            pthread_mutex_lock(&lock);
            timerProc = NULL;
            locked = false;
            pthread_cond_broadcast(&cond);

            if (cur->that)
            {
                // 表示有名定时器
                // 如果运行，并且是重复定时器，就重新添加到timerHead队列
                // 否则立马回收
                if (cur->that->mIsRunning
                        && cur->that->mRepeat)
                {
                    cur->tiggerTime = timeUs() + ((int64_t)(cur->that->mMSec) * 1000);

                    //printf("tiggerTime: %lld, mSec: %d, timer: %p\n", cur->tiggerTime, cur->that->mMSec, cur->that);
                    _appendTimer(cur);
                } else
                {
                    gPool->recycle(cur);
                }
            } else
            {
                //GEN_Printf(LOG_DUMP, "single slot: %p, that: %p", cur->nodeHandler, cur->that);
                // 说明这是一个事件型的单步定时器处理， singleSlot创建
                if (cur->nodeHandler)
                {
                    // 需要把这个事件节点加到 loopHead队列中， 类似post函数
                    if (loopTail)
                    {
                        loopTail->next = cur->nodeHandler;
                    } else
                    {
                        loopHead = cur->nodeHandler;
                    }
                    loopTail = cur->nodeHandler;
                    interrupted = true;
                }
                // 表示是匿名的单步定时器
                // 直接回收
                gPool->recycle(cur);
            }
        }

    }
}

void QueuePrivate::procEventFunc()
{
    DispatchItem *item = head;
    head = item->next;
    if (!head)
    {
        tail = NULL;
    }
    item->next = NULL;
    // Unlock, because we want to allow other threads to queue items
    // while the dispatch item is processed.
    // At the same time, we must prevent other threads from returning
    // from lock(), which is done by locked=true.
    SF_ASSERT(!locked);
    locked = true;
    pthread_mutex_unlock(&lock);

    (item->fn_data->*(item->fn))();

    pthread_mutex_lock(&lock);
    SF_ASSERT(locked);
    locked = false;
    if (item->asynchronous)
    {
#ifdef USE_DESTRUC_OBJ
        if (item->delobject)
        {
            delete item->fn_data;
        }
#endif // USE_DESTRUC_OBJ
        gPool->recycle(item);
    } else
    {
        item->completed = true;
    }

    // Wakeup run(), also lock().
    pthread_cond_broadcast(&cond);
}

void QueuePrivate::interrupt()
{
    exitLoop = true;
    interrupted = true;
}

void QueuePrivate::runIdle()
{
    // GEN_Printf(LOG_WARN, "----------run idle---------------");
}

Timer::Timer(Queue *q)
    : d(NULL)
    , mIsRunning(false), mRepeat(false)
    , mMSec(0),fn(NULL), ctx(NULL)
{
    d = q->d->ref();
    // GEN_Printf(LOG_DEBUG, "d: %p, q->d: %p", d, q->d);
}

//Timer::Timer(QueuePrivate *q)
//    : d(q)
//    , mIsRunning(false)
//    , mMSec(0),fn(NULL), ctx(NULL)
//{
//    GEN_Printf(LOG_DEBUG, "d: %p, q: %p", d, q);
//}

Timer::~Timer()
{
    d->unref();
}

void Timer::setHandler(Object::Callback1 fn, Object *ctx)
{
    this->fn  = fn;
    this->ctx = ctx;
}

void Timer::restart(int msec)
{
    SF_ASSERT(fn);

    pthread_mutex_lock(&d->lock);

    mMSec      = msec;
    mIsRunning = true;

    ItemTimer *item = d->findTimer(this);
    if (item)
    {
        item->fn         = this->fn;
        item->fn_data    = this->ctx;
        item->tiggerTime = timeUs() + msec * 1000;
        item->that       = this;
        item->nodeHandler= nullptr;
        d->_appendTimer(item);
    } else
    {
        d->appendTimer(this, msec * 1000);
    }

    pthread_cond_broadcast(&d->cond);
    pthread_mutex_unlock(&d->lock);

    d->wakeup();
}

void Timer::stop()
{    
    pthread_mutex_lock(&d->lock);
    mIsRunning = false;
    // 如果定时器正在跑就把标志(isRunning)至false,
    // 事件循环就会在定时器处理之后，删除节点
    // 当前表示定时器没有在执行，因此可以直接删除节点
    // GEN_Printf(LOG_DEBUG, "   d: %p, d->timerProc: %d", d, d->timerProc);
    if (this != d->timerProc)
    {
        d->removeTimer(this);
    }

    pthread_mutex_unlock(&d->lock);
}

Loop::Loop()
    : d(NULL)
    , q(NULL)
{
}

Loop::~Loop()
{
    SF_ASSERT(q && d);
    q->run(FUNC(&QueuePrivate::interrupt), d);
    d->unref();
}

void Loop::setQueue(Queue *queue)
{
    q = queue;
    d = q->d->ref();
}

void Loop::post(const Loop::Evt &e)
{
    LoopEvtItem *item = gPool->getLoopNode(e);

    pthread_mutex_lock(&d->lock);
    if (d->loopTail)
    {
        d->loopTail->next = item;
    } else
    {
        d->loopHead = item;
    }
    d->loopTail = item;
    d->interrupted = true;
    // GEN_Printf(LOG_DUMP, "Loop Post: %p", item);
    pthread_cond_broadcast(&d->cond);
    pthread_mutex_unlock(&d->lock);

    d->wakeup();
}

void Loop::postFIFO(const Loop::Evt &e)
{
    LoopEvtItem *item = gPool->getLoopNode(e);

    pthread_mutex_lock(&d->lock);
    if (d->loopHead)
    {
        item->next = d->loopHead->next;
    } else
    {
        item->next = NULL;
        d->loopTail = item;
    }

    // GEN_Printf(LOG_DUMP, "Loop Post: %p", item);
    d->loopHead = item;
    d->interrupted = true;
    pthread_cond_broadcast(&d->cond);
    pthread_mutex_unlock(&d->lock);

    d->wakeup();
}

void Loop::exit()
{
    q->run(FUNC(&QueuePrivate::interrupt), d);
}

void Loop::singleSlot(int msec, Object *ctx, Object::Callback1 fn)
{
    pthread_mutex_lock(&d->lock);
    ItemTimer *item  = gPool->getNode<ItemTimer>(fn, ctx, true);
    item->tiggerTime = timeUs() + msec * 1000;
    item->that       = NULL;
    d->_appendTimer(item);
    pthread_cond_broadcast(&d->cond);
    pthread_mutex_unlock(&d->lock);

    d->wakeup();
}

void Loop::singleSlot(int msec, const Loop::Evt &evt)
{
    pthread_mutex_lock(&d->lock);

//    ItemTimer *cur = d->timerhead;
//    while (cur)
//    {
//        GEN_Printf(LOG_DEBUG, "Timer[%p] Info, tigger: %lld, handler: %p, fndata: %p", cur,
//                   (long long)d->timerhead->tiggerTime, cur->that, cur->fn_data);
//        cur = (ItemTimer *)cur->next;
//    }

    ItemTimer *item  = gPool->getNode<ItemTimer>(FUNC(&QueuePrivate::runIdle), d, true);
    item->tiggerTime = timeUs() + msec * 1000;
    item->that       = NULL;
    item->nodeHandler= gPool->getLoopNode(evt);
    d->_appendTimer(item);

//    GEN_Printf(LOG_WARN, "------------------------------------");
//    cur = d->timerhead;
//    while (cur)
//    {
//        GEN_Printf(LOG_DEBUG, "Timer[%p] Info, tigger: %lld, handler: %p, fndata: %p", cur,
//                   (long long)d->timerhead->tiggerTime, cur->that, cur->fn_data);
//        cur = (ItemTimer *)cur->next;
//    }

    pthread_cond_broadcast(&d->cond);
    pthread_mutex_unlock(&d->lock);

    d->wakeup();
}

void Loop::loopBreak()
{
    d->interrupt();
}

void Loop::exec()
{
    SF_ASSERT(q && d);

    d->exitLoop = false;

    while (!d->exitLoop)
    {
        pthread_mutex_lock(&d->lock);

        while (d->loopHead)
        {
            LoopEvtItem *item = d->loopHead;
            d->loopHead = item->next;
            if (!d->loopHead)
            {
                d->loopTail = NULL;
            }
            item->next = NULL;
            d->locked = true;
            pthread_mutex_unlock(&d->lock);

            procSubMsg(item->evt);
            //GEN_Printf(LOG_DUMP, "Loop Proc: %p", item);
            gPool->recycle(item);

            pthread_mutex_lock(&d->lock);
            d->locked = false;
        }

        pthread_mutex_unlock(&d->lock);

        q->queueProcess();
    }
}

void Loop::execOnce(int msec)
{
    pthread_mutex_lock(&d->lock);

    while (d->loopHead)
    {
        LoopEvtItem *item = d->loopHead;
        d->loopHead = item->next;
        if (!d->loopHead)
        {
            d->loopTail = NULL;
        }
        item->next = NULL;
        d->locked = true;
        pthread_mutex_unlock(&d->lock);

        procSubMsg(item->evt);
        //GEN_Printf(LOG_DUMP, "Loop Proc: %p", item);
        gPool->recycle(item);

        pthread_mutex_lock(&d->lock);
        d->locked = false;
    }

    pthread_mutex_unlock(&d->lock);

    q->queueProcess(msec);
}

void dispatch::Timer::setMSec(int mSec)
{
    mMSec = mSec;
}

void dispatch::Timer::setRepeat(bool value)
{
    mRepeat = value;
}

bool dispatch::Timer::isRunning() const
{
    return mIsRunning;
}

bool dispatch::Timer::repeat() const
{
    return mRepeat;
}

int dispatch::Timer::mSec() const
{
    return mMSec;
}

}
